package com.bangcommunity.bbframe.fdfs;

import com.bangcommunity.bbframe.common.exception.ValidateUtil;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.fdfs.config.FdfsClientConfig;
import com.bangcommunity.bbframe.fdfs.connection.StorageConnection;
import com.bangcommunity.bbframe.fdfs.connection.TrackerConnection;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
import org.csource.common.NameValuePair;
import org.csource.fastdfs.DownloadStream;
import org.csource.fastdfs.FileInfo;
import org.csource.fastdfs.ProtoCommon;
import org.csource.fastdfs.StorageClient;
import org.csource.fastdfs.TrackerGroup;
import org.csource.fastdfs.UploadStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.io.OutputStream;

public class FdfsClientImpl implements FdfsClient {

    private static Logger logger = LoggerFactory.getLogger(FdfsClientImpl.class);
    private GenericKeyedObjectPool<String, TrackerConnection> trackerClientPool;
    private GenericKeyedObjectPool<StorageAddress, StorageConnection> storageClientPool;
    private TrackerGroup trackerGroup;
    private FdfsClientTemplate fdfsClientTemplate;
    private int blockBufferSize = FdfsClientConfig.BLOCK_BUFFER_SIZE_DEFAULT;

    public FdfsClientImpl(TrackerGroup trackerGroup,
            GenericKeyedObjectPool<String, TrackerConnection> trackerClientPool,
            GenericKeyedObjectPool<StorageAddress, StorageConnection> storageClientPool) {
        this(trackerGroup, trackerClientPool, storageClientPool, FdfsClientConfig.BLOCK_BUFFER_SIZE_DEFAULT);
    }

    public FdfsClientImpl(TrackerGroup trackerGroup,
            GenericKeyedObjectPool<String, TrackerConnection> trackerClientPool,
            GenericKeyedObjectPool<StorageAddress, StorageConnection> storageClientPool, int blockBufferSize) {
        super();
        this.trackerGroup = trackerGroup;
        this.trackerClientPool = trackerClientPool;
        this.storageClientPool = storageClientPool;
        this.fdfsClientTemplate = new FdfsClientTemplate(trackerGroup, trackerClientPool, storageClientPool);
        this.blockBufferSize = blockBufferSize;
    }

    @Override
    public void close() {
        this.trackerClientPool.close();
        this.storageClientPool.close();
    }

    @Override
    public FdfsBlockResult<String> upload(InputStream inputStream, String group, String ext, boolean canAppender)
            throws Exception {

        return executeBlockProcess(inputStream, new BlockProcess<String>() {
            @Override
            public String processOnlyOneBlock(byte[] blockBuffer, int readedLength) throws Exception {
                return FdfsClientImpl.this.upload(blockBuffer, 0, readedLength, group, ext, canAppender);
            }

            @Override
            public String processFirstMoreThanOne(byte[] blockBuffer, int readedLength) throws Exception {
                return FdfsClientImpl.this.upload(blockBuffer, group, ext, true);
            }

            @Override
            public String processOther(int i, String firstResult, byte[] blockBuffer, int readedLength)
                    throws Exception {
                FdfsClientImpl.this.append(firstResult, blockBuffer, 0, readedLength);
                return null;
            }
        });
    }

    interface BlockProcess<T> {
        T processOnlyOneBlock(byte[] blockBuffer, int readedLength) throws Exception;

        T processFirstMoreThanOne(byte[] blockBuffer, int readedLength) throws Exception;

        T processOther(int i, T firstResult, byte[] blockBuffer, int readedLength) throws Exception;

    }

    public <T> FdfsBlockResult<T> executeBlockProcess(InputStream inputStream, BlockProcess<T> blockProcess)
            throws Exception {
        int acualBlockSize = this.blockBufferSize * 1024;
        byte[] blockBuffer = new byte[acualBlockSize];
        int readedLength = 0;
        T fullFildId = null;
        int i = 0;
        long total = 0;
        while ((readedLength = inputStream.read(blockBuffer)) > 0) {
            total += readedLength;
            if (null == fullFildId) {
//                有可能一次读取不能读满buffer,不能作为已经读完的条件
//                if (readedLength < acualBlockSize) {
//                    fullFildId = blockProcess.processOnlyOneBlock(blockBuffer,readedLength);
//                    return new FdfsBlockResult(fullFildId, readedLength);
//                }
                fullFildId = blockProcess.processFirstMoreThanOne(blockBuffer, readedLength);
            } else {
                blockProcess.processOther(i, fullFildId, blockBuffer, readedLength);
            }
            i++;
        }
        return new FdfsBlockResult<>(fullFildId, total);
    }

    @Override
    public String upload(InputStream inputStream, long fileSize, String group, String ext, boolean canAppender)
            throws Exception {
        ValidateUtil.isTrue(fileSize > 0, "文件大小必须大于0");
        return fdfsClientTemplate.executeWritable(group, (grp, fid, storageConnection) -> {
            String[] results = storageConnection.doUploadFile(
                    canAppender ? ProtoCommon.STORAGE_PROTO_CMD_UPLOAD_APPENDER_FILE
                            : ProtoCommon.STORAGE_PROTO_CMD_UPLOAD_FILE,
                    group, null, null, ext, fileSize, new UploadStream(inputStream, fileSize));
            logger.info("upload result:{}", StringUtils.join(results));
            ValidateUtil.isTrue(null != results && results.length == 2, "上传文件失败");
            return results[0] + SPLIT_GROUP_NAME_AND_FILENAME_SEPERATOR + results[1];
        });

    }

    @Override
    public String upload(byte[] fileBuff, String group, String ext, boolean canAppender) throws Exception {
        return upload(fileBuff, 0, fileBuff.length, group, ext, canAppender);
    }

    @Override
    public String upload(byte[] fileBuff, int offset, int length, String group, String ext, boolean canAppender)
            throws Exception {
        return fdfsClientTemplate.executeWritable(group, (grp, fid, storageConnection) -> {
            String[] results = storageConnection.doUploadFile(
                    canAppender ? ProtoCommon.STORAGE_PROTO_CMD_UPLOAD_APPENDER_FILE
                            : ProtoCommon.STORAGE_PROTO_CMD_UPLOAD_FILE,
                    group, null, null, ext, length, new StorageClient.UploadBuff(fileBuff, offset, length));
            ValidateUtil.isTrue(null != results && results.length == 2, "上传文件失败");
            return results[0] + SPLIT_GROUP_NAME_AND_FILENAME_SEPERATOR + results[1];
        });
    }

    @Override
    public boolean append(String fullFileId, byte[] fileBuff) throws Exception {
        return append(fullFileId, fileBuff, 0, fileBuff.length);
    }

    @Override
    public boolean append(String fullFileId, byte[] fileBuff, int offset, int length) throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId, (grp, fid, storageConnection) -> (0 == storageConnection
                .doAppendFile(grp, fid, length, new StorageClient.UploadBuff(fileBuff, offset, length))));
    }

    @Override
    public FdfsBlockResult<Boolean> append(String fullFileId, InputStream inputStream) throws Exception {
        return executeBlockProcess(inputStream, new BlockProcess<Boolean>() {
            @Override
            public Boolean processOnlyOneBlock(byte[] blockBuffer, int readedLength) throws Exception {
                return FdfsClientImpl.this.append(fullFileId, blockBuffer, 0, readedLength);
            }

            @Override
            public Boolean processFirstMoreThanOne(byte[] blockBuffer, int readedLength) throws Exception {
                return processOnlyOneBlock(blockBuffer, readedLength);
            }

            @Override
            public Boolean processOther(int i, Boolean firstResult, byte[] blockBuffer, int readedLength)
                    throws Exception {
                return processOnlyOneBlock(blockBuffer, readedLength);
            }
        });
    }

    @Override
    public boolean append(String fullFileId, InputStream inputStream, long fileSize) throws Exception {

        return fdfsClientTemplate.executeUpdatable(fullFileId, (grp, fid, storageConnection) -> (0 == storageConnection
                .doAppendFile(grp, fid, fileSize, new UploadStream(inputStream, fileSize))));
    }

    @Override
    public byte[] download(String fullFileId) throws Exception {
        return download(fullFileId, 0, 0);
    }

    @Override
    public boolean download(String fullFileId, OutputStream outputStream) throws Exception {
        return download(fullFileId, 0, 0, outputStream);
    }

    @Override
    public byte[] download(String fullFileId, long fileOffset, long downloadBytes) throws Exception {
        return fdfsClientTemplate.executeReadable(fullFileId,
                (grp, fid, storageConnection) -> storageConnection.downloadFile(grp, fid, fileOffset, downloadBytes));
    }

    @Override
    public boolean download(String fullFileId, long fileOffset, long downloadBytes, OutputStream outputStream)
            throws Exception {
        return fdfsClientTemplate.executeReadable(fullFileId, (grp, fid, storageConnection) -> (0 == storageConnection
                .downloadFile(grp, fid, fileOffset, downloadBytes, new DownloadStream(outputStream))));
    }

    @Override
    public FdfsSourceInputStream getFdfsSourceInputStream(String fullFileId, long fileOffset,
            long downloadBytes) throws Exception {
        return fdfsClientTemplate.executeReadable(fullFileId, (grp, fid, storageConnection) ->  storageConnection
                .getFdfsSourceInputStream(grp, fid, fileOffset, downloadBytes));
    }

    @Override
    public boolean modify(String fullFileId, byte[] fileBuff) throws Exception {
        return modify(fullFileId, 0, fileBuff);
    }

    @Override
    public FdfsBlockResult<Boolean> modify(String fullFileId, InputStream inputStream) throws Exception {
        return modify(fullFileId, 0, inputStream);
    }

    @Override
    public boolean modify(String fullFileId, InputStream inputStream, long fileSize) throws Exception {
        return modify(fullFileId, 0, inputStream, fileSize);
    }

    @Override
    public boolean modify(String fullFileId, long fileOffset, byte[] fileBuff, int offset, int length)
            throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId, (grp, fid, storageConnection) -> (0 == storageConnection
                .doModifyFile(grp, fid, fileOffset, length, new StorageClient.UploadBuff(fileBuff, offset, length))));
    }

    @Override
    public boolean modify(String fullFileId, long fileOffset, byte[] fileBuff) throws Exception {
        return modify(fullFileId, fileOffset, fileBuff, 0, fileBuff.length);
    }

    @Override
    public boolean modify(String fullFileId, long fileOffset, InputStream inputStream, long fileSize) throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId, (grp, fid, storageConnection) -> (0 == storageConnection
                .doModifyFile(grp, fid, fileOffset, fileSize, new UploadStream(inputStream, fileSize))));
    }

    @Override
    public FdfsBlockResult<Boolean> modify(String fullFileId, long fileOffset, InputStream inputStream)
            throws Exception {
        return executeBlockProcess(inputStream, new BlockProcess<Boolean>() {
            @Override
            public Boolean processOnlyOneBlock(byte[] blockBuffer, int readedLength) throws Exception {
                return FdfsClientImpl.this.modify(fullFileId, fileOffset, blockBuffer, 0, readedLength);
            }

            @Override
            public Boolean processFirstMoreThanOne(byte[] blockBuffer, int readedLength) throws Exception {
                return FdfsClientImpl.this.modify(fullFileId, fileOffset, blockBuffer, 0, readedLength);
            }

            @Override
            public Boolean processOther(int i, Boolean firstResult, byte[] blockBuffer, int readedLength)
                    throws Exception {
                return FdfsClientImpl.this.modify(fullFileId, fileOffset + i * blockBuffer.length, blockBuffer, 0,
                        readedLength);
            }
        });
    }

    @Override
    public FileInfo getFileInfo(String fullFileId) throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId,
                (grp, fid, storageConnection) -> storageConnection.getFileInfo(grp, fid));
    }

    @Override
    public boolean setMetadata(String fullFileId, NameValuePair[] metaList, boolean isMerge) throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId,
                (grp, fid,
                        storageConnection) -> (0 == storageConnection.setMetadata(grp, fid, metaList,
                                isMerge ? ProtoCommon.STORAGE_SET_METADATA_FLAG_MERGE
                                        : ProtoCommon.STORAGE_SET_METADATA_FLAG_OVERWRITE)));
    }

    @Override
    public NameValuePair[] getMetadata(String fullFileId) throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId,
                (grp, fid, storageConnection) -> storageConnection.getMetadata(grp, fid));
    }

    @Override
    public boolean delete(String fullFileId) throws Exception {
        return fdfsClientTemplate.executeUpdatable(fullFileId,
                (grp, fid, storageConnection) -> (0 == storageConnection.deleteFile(grp, fid)));
    }

    public GenericKeyedObjectPool<StorageAddress, StorageConnection> getStorageClientPool() {
        return storageClientPool;
    }

    public void setStorageClientPool(GenericKeyedObjectPool<StorageAddress, StorageConnection> storageClientPool) {
        this.storageClientPool = storageClientPool;
    }

    public GenericKeyedObjectPool<String, TrackerConnection> getTrackerClientPool() {
        return trackerClientPool;
    }

    public void setTrackerClientPool(GenericKeyedObjectPool<String, TrackerConnection> trackerClientPool) {
        this.trackerClientPool = trackerClientPool;
    }

    public TrackerGroup getTrackerGroup() {
        return trackerGroup;
    }

    public void setTrackerGroup(TrackerGroup trackerGroup) {
        this.trackerGroup = this.trackerGroup;
    }

    public FdfsClientTemplate getFdfsClientTemplate() {
        return fdfsClientTemplate;
    }

    public void setFdfsClientTemplate(FdfsClientTemplate fdfsClientTemplate) {
        this.fdfsClientTemplate = fdfsClientTemplate;
    }

    public int getBlockBufferSize() {
        return blockBufferSize;
    }

    public void setBlockBufferSize(int blockBufferSize) {
        this.blockBufferSize = blockBufferSize;
    }

}
